package com.ld.producer.serializer;

import org.apache.kafka.common.serialization.Serializer;

import java.nio.ByteBuffer;
import java.util.Map;

/**
 * @author ld
 *
 * 自定义一个序列化器
 */
public class MySerializer implements Serializer<Person> {

    /**
     * 配置参数信息，在创建 KafkaProducer 实例时被调用
     * @param map
     * @param b
     */
    @Override
    public void configure(Map<String, ?> map, boolean b) {}

    /**
     * 用于实现序列化逻辑
     * @param topic 主题
     * @param person 数据
     * @return
     */
    @Override
    public byte[] serialize(String topic, Person person) {
        if(person == null){
            return null;
        }

        byte[] name, age;
        try{
            if(person.getName() != null){
                name = person.getName().getBytes("UTF-8");
            }else {
                name = new byte[0];
            }

            if(person.getAge() > 0){
                age = String.valueOf(person.getAge()).getBytes("UTF-8");
            }else {
                age = new byte[0];
            }

            ByteBuffer buffer = ByteBuffer.allocate(4 + 4 + name.length + age.length);
            buffer.putInt(name.length);
            buffer.put(name);
            buffer.putInt(age.length);
            buffer.put(age);
            return buffer.array();
        }catch (Exception e){
            e.printStackTrace();
        }
        return new byte[0];
    }

    /**
     * 关闭序列化器时使用
     */
    @Override
    public void close() {
        System.out.println("关闭序列化器......");
    }
}
